package top.wintp.offlinedataanalysis.anlyser.mr.nu;

import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.List;

import cn.hutool.core.convert.Convert;
import top.wintp.offlinedataanalysis.anlyser.dim.StatsCommonDimension;
import top.wintp.offlinedataanalysis.anlyser.dim.StatsUserDimension;
import top.wintp.offlinedataanalysis.anlyser.dim.base.BrowserDimension;
import top.wintp.offlinedataanalysis.anlyser.dim.base.DateDimension;
import top.wintp.offlinedataanalysis.anlyser.dim.base.KpiDimension;
import top.wintp.offlinedataanalysis.anlyser.dim.base.PlatformDimension;
import top.wintp.offlinedataanalysis.anlyser.value.map.TimeOutputValue;
import top.wintp.offlinedataanalysis.common.DateEnum;
import top.wintp.offlinedataanalysis.common.EventLogConstant;
import top.wintp.offlinedataanalysis.common.KpiType;

/**
 * @author: pyfysf
 * <p>
 * @qq: 337081267
 * <p>
 * @CSDN: http://blog.csdn.net/pyfysf
 * <p>
 * @blog: http://wintp.top
 * <p>
 * @email: pyfysf@163.com
 * <p>
 * @time: 2019/9/5
 */
public class NewInstallUserMapper extends TableMapper<StatsUserDimension, TimeOutputValue> {
    private static final Logger logger = LoggerFactory.getLogger(NewInstallUserMapper.class);

    /**
     * 初始化输出的key
     */
    private StatsUserDimension userDimension = new StatsUserDimension();

    /**
     * 初始化输出的value
     */
    private TimeOutputValue timeOutputValue = new TimeOutputValue();

    /**
     * HBase 表的列族
     */
    private byte[] tableFamilyName = Bytes.toBytes(EventLogConstant.EVENT_LOGS_FAMILY_NAME);

    /**
     * 新增用户的模块 KPI
     */
    private String newInstallUserKpiType = KpiType.NEW_INSTALL_USER.name;

    /**
     * 分析浏览器模块的KPI  以浏览器维度进行分析 某款浏览器的某个版本下 新增用户的个数
     */
    private String browserUserKpiType = KpiType.BROWSER_NEW_INSTALL_USER.name;

    @Override
    protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
        //    获取HBase中的数据
        //    服务器时间
        String serviceTime = Bytes.toString(CellUtil.cloneValue(value.getColumnLatestCell(tableFamilyName, Bytes.toBytes(EventLogConstant.LOG_COLUMN_NAME_SERVER_TIME))));
        //    浏览器名称
        String browserName = Bytes.toString(CellUtil.cloneValue(value.getColumnLatestCell(tableFamilyName, Bytes.toBytes(EventLogConstant.LOG_COLUMN_NAME_BROWSER_NAME))));
        //    浏览器版本
        String browserVersion = Bytes.toString(CellUtil.cloneValue(value.getColumnLatestCell(tableFamilyName, Bytes.toBytes(EventLogConstant.LOG_COLUMN_NAME_BROWSER_VERSION))));
        //    uuid
        String uuid = Bytes.toString(CellUtil.cloneValue(value.getColumnLatestCell(tableFamilyName, Bytes.toBytes(EventLogConstant.LOG_COLUMN_NAME_UUID))));
        //    platform
        String platform = Bytes.toString(CellUtil.cloneValue(value.getColumnLatestCell(tableFamilyName, Bytes.toBytes(EventLogConstant.LOG_COLUMN_NAME_PLATFORM))));

        //    封装单一维度
        //    日期维度
        DateDimension dateDimension = DateDimension.buildDate(Convert.toLong(serviceTime), DateEnum.DAY);
        //    浏览器维度
        List<BrowserDimension> browserDimensions = BrowserDimension.buildList(browserName, browserVersion);
        //    平台维度
        List<PlatformDimension> platformDimensions = PlatformDimension.buildList(platform);

        //    填充输出的value
        timeOutputValue.setId(uuid);
        timeOutputValue.setTime(Convert.toLong(serviceTime));

        //通用维度
        StatsCommonDimension statsCommonDimension = new StatsCommonDimension();
        //由于基本用户信息分析 新增用户没有浏览器信息 创建一个空的浏览器信息的维度
        BrowserDimension defaultBrowserDimension = new BrowserDimension("", "");

        //    kpi维度 这个是为了声明分析的是哪个模块 方便统一分析某些相同逻辑的数据 最后向哪一张mysql表进行插入
        for (PlatformDimension platformDimension : platformDimensions) {
            //用户基本信息 新增用户
            KpiDimension kpiDimension = new KpiDimension(newInstallUserKpiType);
            statsCommonDimension.setKpi(kpiDimension);
            statsCommonDimension.setDate(dateDimension);
            userDimension.setBrowser(defaultBrowserDimension);
            statsCommonDimension.setPlatform(platformDimension);

            //    组合用户基本信息之新增用户的维度
            userDimension.setStatsCommon(statsCommonDimension);
            //    数据写出
            context.write(userDimension, timeOutputValue);

            //组合浏览器信息 这个循环其实就是为分析浏览器之新增用户使用的
            for (BrowserDimension browserDimension : browserDimensions) {
                kpiDimension.setKpiName(browserUserKpiType);
                userDimension.setBrowser(browserDimension);
                context.write(userDimension, timeOutputValue);
            }
        }

    }
}
